package gridy.fitness;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.Semaphore;

import org.apache.axiom.om.OMAbstractFactory;
import org.apache.axiom.om.OMElement;
import org.apache.axiom.om.OMFactory;
import org.apache.axiom.om.OMNamespace;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.client.Options;
import org.apache.axis2.client.ServiceClient;
import org.apache.log4j.Logger;
import org.osoa.sca.annotations.EagerInit;
import org.osoa.sca.annotations.Scope;

@EagerInit
@Scope("COMPOSITE")
public class FitnessCalculatorImpl implements FitnessCalculator, FitnessCalculatorMaster {
    
    static Logger log = Logger.getLogger(FitnessCalculatorImpl.class);

    private Set<String> workers = new HashSet<String>();
    private Semaphore semaphore;
    private double[] gResult;
    
    public FitnessCalculatorImpl() {
        log.info("Tworze FitnessCalculatorImpl");
    }
    
    @Override
    public double[] calculate(double[] arguments, int numOfProblems) {
        
        
        assert arguments.length % numOfProblems == 0;
        int problemSize = arguments.length / numOfProblems;
        int numOfWorkers = workers.size();
        
        if (workers.size() > 0) {

            this.gResult = new double[numOfProblems];
            int problemsPerWorker = (int)(numOfProblems / numOfWorkers);
            int additionalProblems = numOfProblems % numOfWorkers;

            log.debug(String.format(
                    "arguments.length = %d, numOfProblems = %d, problemSize = %d, numOfWorkers = %d, problemsPerWorker = %d",
                    arguments.length,numOfProblems,problemSize,numOfWorkers,problemsPerWorker));

            semaphore = new Semaphore(1 - numOfWorkers);

            Iterator<String> workersIt = workers.iterator();
            int problemBegIndex = 0;
            for (int workerNum = 0; workerNum < numOfWorkers; workerNum++) {
                String workerName = workersIt.next();

                double[] job;
                int currentNumOfProblems;
                if ( (workerNum != (numOfWorkers - 1)) || (additionalProblems == 0)) {

                    job = new double[problemsPerWorker * problemSize];
                    currentNumOfProblems = problemsPerWorker;
                }
                else {
                    // do ostatniego wysyla sie pozostale zadania
                    job = new double[(problemsPerWorker + additionalProblems) * problemSize];
                    currentNumOfProblems = problemsPerWorker+additionalProblems;
                }

                for (int i = 0; i < job.length; i++) {
                    job[i] = arguments[(problemBegIndex * problemSize) + i];
                }
                send(workerName, job, currentNumOfProblems, problemBegIndex);

                problemBegIndex += problemsPerWorker;
            }
            try {
                // czeka aż workerzy zwrócą wyniki
                semaphore.acquire();
            }
            catch (InterruptedException e) {
                e.printStackTrace();
                System.exit(1);
            }
            assert isResultCorrect(gResult, numOfProblems, problemSize, arguments,0.0001);
        }
        else {
            gResult = calculateLocally(numOfProblems, problemSize, arguments);
            assert isResultCorrect(gResult, numOfProblems, problemSize, arguments,0.0001);
        }
        
        return gResult;
    }
    
    private double[] calculateLocally(int numOfProblems, int problemSize,
            double[] arguments) {
        
        double[] result = new double[numOfProblems];
        
        for (int i = 0; i < numOfProblems; i++) {
            double[] subarray = new double[problemSize];
            for (int j = 0; j < subarray.length; j++) {
                subarray[j] = arguments[i * problemSize + j];
            }
            
            for (int j = 0; j < 100; j++) {
                result[i] = getValue(subarray);
            }
        }
        
        return result;
    }

    private boolean isResultCorrect(double[] result, int numOfProblems, int problemSize, double[] arguments, double epsilon) {
        
        double[] locallyCalculatedResult = calculateLocally(numOfProblems, problemSize, arguments);
        
        for (int i = 0; i < locallyCalculatedResult.length; i++) {
            if (Math.abs(result[i] - locallyCalculatedResult[i]) > epsilon) {
                return false;
            }
        }
        return true;
    }
    
    private double getValue(double[] genes) {
        double value = 10.0 * genes.length;
        for (int i = 0; i < genes.length; i++) {
            value = value + Math.pow(genes[i], 2.0) - 10.0 * Math.cos(2.0 * Math.PI * genes[i]);
        }
        return value;
    }

    private void send(String workerAddress, double[] arguments,
            int numOfProblems, int firstProblemIndex) {
        
        log.debug(String.format(
                "Wysylam do %s, arguments.length = %d, numOfProblems = %d, firstProblemIndex = %d",
                workerAddress, arguments.length, numOfProblems, firstProblemIndex));
        
        try {
            
            ServiceClient sc = new ServiceClient();
            Options opts = new Options();
            opts.setTo(new EndpointReference("http://"+workerAddress+"/FitnessCalculatorWorkerService"));
            sc.setOptions(opts);
            
            OMFactory fac = OMAbstractFactory.getOMFactory();
            OMNamespace omNs = fac.createOMNamespace(
                    "http://ws.apache.org/axis2/xsd", "ns1");
            OMElement payload = fac.createOMElement("calculate", omNs);
            OMElement value;
            for (double arg : arguments) {
                value = fac.createOMElement("arg0", omNs);
                value.setText(String.valueOf(arg));
                payload.addChild(value);
            }
            value = fac.createOMElement("arg1", omNs);
            value.setText(String.valueOf(numOfProblems));
            payload.addChild(value);
            
            value = fac.createOMElement("arg2", omNs);
            value.setText(String.valueOf(firstProblemIndex));
            payload.addChild(value);
                
            sc.fireAndForget(payload);
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void register(String workerAddress) {
        
        log.info("Worker '"+workerAddress+"' zarejestrowal sie");
        for (String worker : workers) {
            log.info("\tByl juz worker: "+worker);
        }
        workers.add(workerAddress);
    }

    @Override
    public void unregister(String workerAddress) {
        
        log.info("Worker '"+workerAddress+"' wyrejestrowal sie");
        workers.remove(workerAddress);
        for (String worker : workers) {
            log.info("\tPozostali jeszcze: "+worker);
        }
    }

    @Override
    synchronized public void putResults(double[] result, int firstProblemIndex) {
        
        log.debug(String.format("Otrzymalem wyniki, result.length = %d, firstProblemIndex = %d",
                result.length, firstProblemIndex));
        
        for (int i = 0; i < result.length; i++) {
            this.gResult[firstProblemIndex+i] = result[i];
        }
        semaphore.release();
    }
}
